package cn.itcast.streaming.utils.test;

import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.IterativeCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.io.Serializable;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

/**
 * Author itcast
 * Date 2021/9/28 9:17
 * Desc TODO
 */
public class LoginFailDemo {
    public static void main(String[] args) throws Exception {
        //1：初始化流式运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //2：设置并行度为1
        env.setParallelism(1);

        //3：指定数据按照事件时间进行处理
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //4：构建数据源
        DataStream<LoginEvent> loginEventStream = env.fromCollection(Arrays.asList(
                new LoginEvent(1, "192.168.0.1", "fail", 1558430842000L),   //2019-05-21 17:27:22
                new LoginEvent(1, "192.168.0.2", "fail", 1558430843000L),   //2019-05-21 17:27:23
                new LoginEvent(1, "192.168.0.3", "fail", 1558430844000L),   //2019-05-21 17:27:24
                new LoginEvent(2, "192.168.10.10", "success", 1558430845000L) //2019-05-21 17:27:25
        )).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<LoginEvent>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(LoginEvent element) {
                return element.getEventTime();
            }
        });

        /**
         * CEP的是由数据和规则模型组成
         * 以mysql举例：
         *      数据是存在于表中的， 也就是静态的数据
         *      查询操作可以是任意的sql语句，任意编写，也就是查询是动态
         * 对于FlinkCEP来说，正好相反：
         *      数据是源源不断的产生的， 也就是动态的数据
         *      查询操作就是定义的规则模型，规则模型一旦被定义，不能修改，如果修改的话必须要重启作业
         * 5：数据的处理
         *  5.1：定义规则模型
         *  5.2：将规则模型应用到数据流中
         *  5.3：获取到符合规则模型的数据
         *
         * 在FlinkCEP中规则模型的定义，需要有三种状态：起始状态（state）->中间状态（state）->结束状态（state），每个状态都可以有一个条件或者多个条件，或者没条件
         * 可以将符合这个条件的数据放到状态中
         * begin：起始状态
         * next、fllowerby：中间状态
         * end：结束状态
         */
        //5.1：定义规则模型
        Pattern<LoginEvent, LoginEvent> loginEventPattern = Pattern.<LoginEvent>begin("begin")
                .where(new IterativeCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent, Context<LoginEvent> context) throws Exception {
                        return loginEvent.getEventType().equals("fail");
                    }
                }) //匹配第一个事件，匹配的是登录失败
                .next("next") //匹配到第一个事件以后，紧跟着一个事件数据，next表示两个事件必须严格的临近
                .where(new IterativeCondition<LoginEvent>() {
                    @Override
                    public boolean filter(LoginEvent loginEvent, Context<LoginEvent> context) throws Exception {
                        return loginEvent.getEventType().equals("fail");
                    }
                }) //匹配第二个事件，匹配的是登录失败
                .within(Time.seconds(3));//定义结束状态，结束状态可以是时间触发也可以是满足了某个事件触发

        //5.2：将规则模型应用到数据流中
        PatternStream<LoginEvent> patternDataStream = CEP
                .pattern(loginEventStream.keyBy(LoginEvent::getUserId), loginEventPattern);

        //5.3：获取到符合规则模型的数据
        /**
         * IN：传入的数据类型
         * OUT：返回值的数据类型
         *    (Long, String, String, Long):(用户id, 登录ip，登录状态，登录时间)
         */
        SingleOutputStreamOperator<Tuple4<Integer, String, String, Long>> loginFailDataStream = patternDataStream
                .select(new PatternSelectFunction<LoginEvent, Tuple4<Integer, String, String, Long>>() {
            //将匹配到的数据查询出来返回
            @Override
            public Tuple4<Integer, String, String, Long> select(Map<String, List<LoginEvent>> pattern) throws Exception {
                //根据刚才的分析，符合规则的数据会存储到状态集合中，也就是state中，所以查找匹配的时候需要在state中获取数据
                LoginEvent loginEvent = pattern.getOrDefault("next", null).iterator().next();

                //返回匹配到的数据
                return Tuple4.of(loginEvent.getUserId(), loginEvent.getIp(),
                        loginEvent.getEventType(), loginEvent.getEventTime());
            }
        });

        //打印出来符合条件的数据
        loginFailDataStream.printToErr("连续两次登录失败的用户>>>");

        //执行任务
        env.execute();
    }

    /**
     * 构建登录对象的样例类
     */
    public static class LoginEvent implements Serializable {
        public int userId;
        public String ip;
        public String eventType;
        public Long eventTime;

        public LoginEvent() {
        }

        public LoginEvent(int userId, String ip, String eventType, Long eventTime) {
            this.userId = userId;
            this.ip = ip;
            this.eventType = eventType;
            this.eventTime = eventTime;
        }

        public int getUserId() {
            return userId;
        }

        public void setUserId(int userId) {
            this.userId = userId;
        }

        public String getIp() {
            return ip;
        }

        public void setIp(String ip) {
            this.ip = ip;
        }

        public String getEventType() {
            return eventType;
        }

        public void setEventType(String eventType) {
            this.eventType = eventType;
        }

        public Long getEventTime() {
            return eventTime;
        }

        public void setEventTime(Long eventTime) {
            this.eventTime = eventTime;
        }

        @Override
        public String toString() {
            return "LoginEvent{" +
                    "userId=" + userId +
                    ", ip='" + ip + '\'' +
                    ", eventType='" + eventType + '\'' +
                    ", eventTime=" + eventTime +
                    '}';
        }
    }
}